package com.caul.demo.hadoop.hdfs.server;

import com.caul.demo.hadoop.hdfs.Cmd;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by BlueDream on 2018-03-27.
 */
public class DataNode {

  private static final ExecutorService executorService = Executors.newFixedThreadPool(10);

  public static void main(String[] args) throws IOException {
    System.out.println("启动DataNode:-------");
    ServerSocket serverSocket = new ServerSocket(Cmd.DATANODE_PORT);
    while (true) {
      Socket socket = serverSocket.accept();
      executorService.submit(new DataNode.Handler(socket));
    }
  }

  private static class Handler<T> implements Callable<T> {

    private Socket socket;

    public Handler(Socket socket) {
      this.socket = socket;
    }

    @Override
    public T call() throws Exception {
      InputStream nnIn = socket.getInputStream();
      OutputStream nnOut = socket.getOutputStream();

      try {
        byte[] bts = new byte[Cmd.DATANODE_CMD_LEN];
        nnIn.read(bts);
        String reqStr = new String(bts).trim();
        System.out.println("客户端請求:" + reqStr);
        String[] cmdAry = reqStr.split(Cmd.SPLIT_CMD);
        String cmd = cmdAry[0];
        String blockId = cmdAry[1];

        switch (cmd) {
          case Cmd.UPLOAD:
            upload(blockId, nnIn, nnOut);
            break;
          case Cmd.DOWNLOAD:
            download(blockId, nnIn, nnOut);
            break;
          default:
            break;
        }

        nnOut.flush();
      } finally {
        nnOut.close();
        nnIn.close();
        socket.close();
      }
      return null;
    }

    /**
     * 上传
     *
     * @return
     */
    private void upload(String blockId, InputStream is, OutputStream os) throws IOException {
      byte[] bts = new byte[4096];
      FileOutputStream fos = new FileOutputStream(new File(Cmd.DATA_PATH + blockId));
      try {
        int read = 0;
        while ((read = is.read(bts)) != -1) {
          fos.write(bts, 0, read);
        }
        fos.flush();
      } finally {
        fos.close();
      }
    }

    /**
     * 下载
     *
     * @return
     */

    private void download(String blockId, InputStream is, OutputStream os) throws IOException {
      FileInputStream fis = new FileInputStream(Cmd.DATA_PATH + blockId);
      try {
        byte[] bts = new byte[4096];
        int read = 0;
        while ((read = fis.read(bts)) != -1) {
          os.write(bts, 0, read);
          os.flush();
        }
      } finally {
        fis.close();
      }
    }
  }

}
